package net.techfuser.bigscreen;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.techfuser.bigscreen.entity.ChannelStoreSummary;
import net.techfuser.bigscreen.entity.ChannelSummary;
import net.techfuser.bigscreen.entity.EntSummary;
import net.techfuser.bigscreen.entity.StoreSummary;
import net.techfuser.bigscreen.sink.QnhBigScreenRedisMapper;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.*;

/**
 * @Description:
 * @Author 左龙龙
 * @Date 21-4-6
 * @Version 1.0
 **/
public class OrderTask {

    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 1.source
        FlinkKafkaConsumer<String> flinkKafkaConsumer = getFlinkKafkaConsumer();
//        flinkKafkaConsumer.setStartFromLatest();
        flinkKafkaConsumer.setStartFromEarliest();
        flinkKafkaConsumer.setStartFromTimestamp(1618844400000L);
        DataStreamSource<String> kafkaDS = env.addSource(flinkKafkaConsumer);


        // 2.transformation
        SingleOutputStreamOperator<JSONObject> orderDS = kafkaDS.map(JSON::parseObject)
                .filter(oplog -> "i".equals(oplog.getString("op")))
                .map(oplog -> oplog.getJSONObject("data"))
//                .filter(order -> {
//                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
//                    return "2021-04-08".equals(sdf.format(order.getDate("order_time")));
//                })
                .filter(order -> order.getDate("order_time") != null)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                                .withTimestampAssigner((order, timestamp) -> order.getDate("order_time").getTime())
                                .withIdleness(Duration.ofSeconds(1))
                ).name("订单过滤转换");

//        orderDS.print("orderDS");


        SingleOutputStreamOperator<ChannelStoreSummary> tempAggregate = orderDS.keyBy(new EntRegionChannelKeySelector())
                .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
                .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(1)))
                .aggregate(new OrderAggregate(), new OrderWindow())
                .name("根据门店渠道分组");
//        tempAggregate.print("tempAggregate");


        SingleOutputStreamOperator<BigScreenModel> resultDS = tempAggregate.keyBy((ChannelStoreSummary row) -> row.timestamp)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
                .apply(new WindowFunction<ChannelStoreSummary, BigScreenModel, String, TimeWindow>() {

                    @Override
                    public void apply(String s, TimeWindow window, Iterable<ChannelStoreSummary> elements, Collector<BigScreenModel> out) throws Exception {
                        BigScreenModel result = new BigScreenModel();
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//                        result.window_time = String.format("%s ~ %s", sdf.format(window.getStart()), sdf.format(window.getEnd()));
                        result.date = sdf.format(new Date());

                        Map<Long, EntSummary> entMap = new HashMap<>();
                        Map<String, ChannelSummary> channelMap = new HashMap<>();
                        Map<String, StoreSummary> storeMap = new HashMap<>();

                        Iterator<ChannelStoreSummary> iterator = elements.iterator();
                        while (iterator.hasNext()) {
                            ChannelStoreSummary row = iterator.next();

                            EntSummary entSummary = entMap.get(row.ent_id);
                            if (null == entSummary) {
                                entSummary = new EntSummary();
                                entSummary.ent_id = row.ent_id;
                                entMap.put(row.ent_id, entSummary);
                            }
                            entSummary.count += row.count;
                            entSummary.total_sale_value += row.total_sale_value;


                            ChannelSummary channelSummary = channelMap.get(row.channel_keyword);
                            if (null == channelSummary) {
                                channelSummary = new ChannelSummary();
                                channelSummary.channel_keyword = row.channel_keyword;
                                channelMap.put(row.channel_keyword, channelSummary);
                            }
                            channelSummary.count += row.count;
                            channelSummary.total_sale_value += row.total_sale_value;


                            String key = String.format("%s_%s", row.ent_id, row.region_code);
                            StoreSummary storeSummary = storeMap.get(key);
                            if (null == storeSummary) {
                                storeSummary = new StoreSummary();
                                storeSummary.ent_id = row.ent_id;
                                storeSummary.region_code = row.region_code;
                                storeSummary.region_name = row.region_name;
                                storeMap.put(key, storeSummary);
                            }
                            storeSummary.count += row.count;
                            storeSummary.total_sale_value += row.total_sale_value;


                            result.total_order_num += row.count;
                            result.total_sale_value += row.total_sale_value;
                            result.window_time = row.window_time;
                        }


                        result.ents = new ArrayList<>(entMap.values());
                        result.channels = new ArrayList<>(channelMap.values());
                        result.stores = new ArrayList<>(storeMap.values());

                        out.collect(result);
                    }
                });



//        resultDS.print("result");


        // 门店销售汇总


        // 3.sink
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder()
                .setHost("192.168.40.221")
                .setPassword("techfuser")
                .setDatabase(6)
                .build();
        RedisSink<BigScreenModel> redisSink = new RedisSink<BigScreenModel>(conf, new QnhBigScreenRedisMapper());
        resultDS.addSink(redisSink).name("数据结果存redis");

        // 4.execute
        env.execute("订单oplog消费");
    }


    private static FlinkKafkaConsumer<String> getFlinkKafkaConsumer() {
        // 准备 kafka 连接参数
        Properties properties = new Properties();
        // 集群地址
        properties.setProperty("bootstrap.servers", "172.16.10.23:9092");
//        properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
        // 消费组
        properties.setProperty("group.id", "qnh-big-screen-test");
        // 有 offset 消费记录则从记录位置继续消息, 没有消费记录则从最新的（最后的）消息开始消费
//        properties.setProperty("auto.offset.reset", "latest");
        // 会开启一个后台线程每隔5秒检查一下 kafka 的分区情况, 实现动态分区检测
//        properties.setProperty("flink.partition-discovery.interval-millis", "5000");
        // 自动提交
        properties.setProperty("enable.auto.commit", "true");
        // 自动提交的时间间隔
        properties.setProperty("auto.commit.interval.ms", "2000");

        return new FlinkKafkaConsumer<>("realtime_order", new SimpleStringSchema(), properties);
    }

}








